/**
 * 
 */
package com.xb.concurrent.hashmap;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 字符统计
 * 
 * @author kexue
 *
 */
public class CharCountTask implements Runnable {
	private Logger logger = LoggerFactory.getLogger(CharCountTask.class);

	private boolean isStop = false;
	private ConcurrentMap<Character, AtomicInteger> charCountMap;
	private BlockingQueue<String> dataQueue;
	private CountDownLatch countDownLatch;

	/**
	 * @param isStop
	 * @param charCountMap
	 */
	public CharCountTask(ConcurrentMap<Character, AtomicInteger> charCountMap, BlockingQueue<String> dataQueue,
			CountDownLatch countDownLatch) {
		this.charCountMap = charCountMap;
		this.dataQueue = dataQueue;
		this.countDownLatch = countDownLatch;
	}

	public void run() {

		try {
			logger.info("test logger");
			while (!isStop) {
				String data = dataQueue.poll();
				if ("END_TASK".equals(data)) {
					logger.info("test END_TASK");
					break;
				} else if (data != null) {
					processData(data);
				} else {
					Thread.currentThread().yield();
				}
			}

			logger.info(" process end");
		} catch (Exception e) {
			// TODO: handle exception
		} finally {
			countDownLatch.countDown();
		}
	}

	/**
	 * @param data
	 */
	private void processData(String data) {
		for (Character ch : data.toCharArray()) {
			AtomicInteger putIfAbsent = charCountMap.putIfAbsent(ch, new AtomicInteger(1));
			if (putIfAbsent != null) {
				// 存在当前key
				AtomicInteger count = charCountMap.get(ch);
				count.incrementAndGet();
			}
		}
	}

	public boolean isStop() {
		return isStop;
	}

	public void setStop(boolean isStop) {
		this.isStop = isStop;
	}

}
